package com.wangyadong.hobby.kafka;


import org.apache.kafka.clients.consumer.ConsumerRecords;

import java.util.Collections;
import java.util.Properties;

/**
 * <p>
 *
 * @author <714037465@qq.com>
 * @since 2018/8/2023:18
 **/
public class KafkaConsumer {
    private static final String KAFKA_INFLUXDB_TOPIC = "event-current-data";

    private static Properties init() {
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "10.10.15.84:9092,10.10.15.85:9092,10.10.15.86:9092");
//        properties.setProperty("bootstrap.servers", "10.10.11.229:9092");
        properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("group.id", "send_demo_group");
        properties.setProperty("client.id", "zhang_test1");
        properties.setProperty("auto.offset.reset", "earliest");
        properties.setProperty("enable.auto.commit", "false");
        return properties;
    }

    public static void main(String[] args) {
        org.apache.kafka.clients.consumer.KafkaConsumer<String, String> cou = new org.apache.kafka.clients.consumer.KafkaConsumer<String, String>(init());
        cou.subscribe(Collections.singletonList(KAFKA_INFLUXDB_TOPIC));//send_demo
        for (; ; ) {
            System.out.println("-------------------->");
            //不要在这处理逻辑
            ConsumerRecords<String, String> records = cou.poll(100);
            records.forEach(recode -> {
                System.out.println(recode.key() + " " + recode.offset() + "  " + recode.value() + "-->" + recode.partition());
            });
        }
    }
}
